package org.zjt.flink;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.*;

/**
 * Description:
 *
 * @author juntao.zhang
 * Date: 2018-09-30 上午10:38
 * @see
 */
public class TableQueryJob {

    public static void main(String[] args) throws Exception {

        // set up execution environment
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env);

        DataSet<WC> input = env.fromElements(new WC("Hello", 1), new WC("Ciao", 1), new WC("Hello", 1));

        // register the DataSet as table "WordCount"
        tEnv.registerDataSet("WordCount", input, "content, cnt");

        // sql 查询
        // run a SQL query on the Table and retrieve the result as a new Table
        Table table = tEnv.sqlQuery("SELECT content, SUM(cnt) as cnt FROM WordCount GROUP BY content ORDER BY cnt DESC");
        DataSet<WC> result = tEnv.toDataSet(table, WC.class);
        result.print();


        // hql 查询
        Table filtered = table.groupBy("content").select("content, cnt.sum as cnt").filter("cnt = 2");
        DataSet<WC> result2 = tEnv.toDataSet(filtered, WC.class);
        result2.print();
    }



    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static final class WC{
        private String content;
        private Integer cnt ;
    }
}
